#include "db/iodatasegment.hxx"
#include "db/datastore_impl.hxx"
#include "db/dbpartition.hxx"
#include "util/serialize.hxx"
#include "util/fs.hxx"
#include "util/io.hxx"
#include "util/common.hxx"
#include "util/lockguard.hxx"
#include "db/metafile.hxx"
#include "db/compact_impl.hxx"
#include "util/logger.hxx"
#include <unistd.h>
#include <stdlib.h>
#include <sstream>



IODataSegmentFile::IODataSegmentFile(const std::string& segFileName, DataSegment* seg)
: m_segFileName(segFileName), 
  m_lastSyncOffset(0), 
  m_changed(false), 
  m_seg(seg) {

}

IODataSegmentFile::~IODataSegmentFile() {

	if(m_raf) { delete m_raf; m_raf = NULL; }
	if(m_writer) { delete m_writer; m_writer = NULL; }
}

DbStatus IODataSegmentFile::init() {

    DbStatus ret;

    int rc = pthread_mutex_init(&m_mutex, NULL);
    if(rc != 0) return DbStatus("IODataSegmentFile::init::pthread_mutex_init", rc);

	bool newfile = false;
	if(!FsUtility::fileExists(m_segFileName)) {
	
		newfile = true;
	}


	m_raf = new PosixRandomAccessFile(m_segFileName);
	//m_raf = new PosixMmapRandomAccessFile(m_segFileName, m_seg->getCapacity() );
    ret = m_raf->init(); if(!ret.isok()) return ret;
	
	m_writer = new PosixMmapSequentialWriter(m_segFileName, getpagesize(),
            DataSegmentHeader::DATASEGMENT_HEADER_LENGTH, true);
	ret = m_writer->init(); if(!ret.isok()) return ret;

	if(newfile) {
	
		ret = saveHeader();
		if(!ret.isok()) return ret;
		m_lastSyncOffset = DataSegmentHeader::DATASEGMENT_HEADER_LENGTH;
	}
	else {
	
		ret = loadHeader();
        if(ret.getCode() == DbStatus::DB_PARTIALREAD || ret.getCode() ==
                DbStatus::DB_CORRUPTION) {
        
            ret = saveHeader();
        }
		if(!ret.isok()) return ret;
		
		m_lastSyncOffset = m_writer->getNextOffset();	
	}


	
	return ret;
}


DbStatus IODataSegmentFile::saveHeader() {

    DbStatus ret;

	std::string header = m_header.serialize();

	ret = m_raf->write(0, header);
	if(!ret.isok()) return ret;

//	ret = m_raf->flush();

	return ret;
}

DbStatus IODataSegmentFile::loadHeader() {

    DbStatus ret;

	std::string header;
	ret = m_raf->read(0, DataSegmentHeader::DATASEGMENT_HEADER_LENGTH, header);
	if(!ret.isok()) return ret;

	m_header.deserialize(header);
	bool isvalid = m_header.checkHeader();
	if(!isvalid) return DbStatus("IODataSegmentFile::loadHeader", DbStatus::DB_CORRUPTION);

	return ret;
}


DbStatus IODataSegmentFile::append(char* buf, uint32_t len) {


    //append may conflict with flush
    LockGuard lock(m_mutex);

	return m_writer->append(buf, len);
}

DbStatus IODataSegmentFile::append(const std::string& data) {

    //append may conflict with flush
    LockGuard lock(m_mutex);

	return m_writer->append(data); 
}

DbStatus IODataSegmentFile::read(uint32_t offset, uint32_t size, std::string& result) {

    DbStatus ret;

    ret = m_raf->read(offset, size, result);
    if(!ret.isok()) return ret;

    if(result.size() != size) ret = DbStatus("IODataSegmentFile::read", DbStatus::DB_CORRUPTION);

    return ret;
}


DbStatus IODataSegmentFile::read(uint32_t offset, 
                                 uint32_t size, 
                                 char* buf,
                                 uint32_t buflen, 
                                 uint32_t& dlen) {

    DbStatus ret;

    ret = m_raf->read(offset, size, buf, buflen, dlen);
    if(!ret.isok()) return ret;

    if(dlen != size) ret = DbStatus("IODataSegmentFile::read", DbStatus::DB_CORRUPTION);

    return ret;
}


DbStatus IODataSegmentFile::flush() {

    DbStatus ret;

    //there is a race condition between flush and compcation/put: If compaction/put
    //set the segment to READ_ONLY after checking the 'if' below, it will crash
    //because m_writer is NULL. To prevent from this, we put a lock in
    //closeWriter
    if(!m_seg->isReadOnly()) {

        LockGuard lock(m_mutex);

        if(m_writer) ret = m_writer->flush();
    }

    //just happen during recovery
    if(m_changed) m_raf->flush();


    return ret;
}


DbStatus IODataSegmentFile::write(uint32_t offset, const std::string& data) {

	return m_raf->write(offset, data);
}


DbStatus IODataSegmentFile::write(uint64_t offset, const char* buf, uint32_t n) {

	return m_raf->write(offset, buf, n);
}

void IODataSegmentFile::closeWriter() {

    LockGuard lock(m_mutex);
    if(m_writer) { delete m_writer; m_writer = NULL; } 
}


DbStatus IODataSegmentFile::dropCache() {

    return m_raf->dropCache();
}

IODataSegment::IODataSegment(uint32_t segId, const std::string& partitionPath, uint32_t capacity)
: DataSegment(segId, partitionPath, capacity), 
  m_segFile(NULL),
  m_notifyCompact(false) {
}


IODataSegment::IODataSegment(uint32_t segId, 
                             DataStoreImpl& ds, 
                             DbPartition& partition, 
                             uint32_t capacity)
: DataSegment(segId, partition.getPartitionPath(), capacity), 
  m_segFile(NULL),
  m_ds(&ds),
  m_partition(&partition),
   m_notifyCompact(false) {

}

IODataSegment::~IODataSegment() {

    MetaFile::getInstance()->removeSeg(m_ds->getDsName(),
                                       m_partition->getPartitionName(),
                                       m_segid);

	if(m_segFile) {delete m_segFile; m_segFile = NULL;}
}

DbStatus IODataSegment::init() {

    //reset data size and alive from metafile
    m_dataSize = MetaFile::getInstance()->getDataSize(m_ds->getDsName(),
                                                      m_partition->getPartitionName(),
                                                      m_segid);

    m_alive = MetaFile::getInstance()->isAlive(m_ds->getDsName(),
                                               m_partition->getPartitionName(),
                                               m_segid);
    std::ostringstream oss;
	oss<<m_partitionPath<<"/"<<m_segid;
	std::string segFileName = oss.str();
	m_segFile = new IODataSegmentFile(segFileName, this);
	
	return m_segFile->init();
}


DbStatus IODataSegment::appendShort(uint16_t value) {

    if(m_segMode == DataSegment::READ_ONLY) 
        return DbStatus("IODataSegment::appendShort", DbStatus::DB_READONLY);

	return m_segFile->append((char*)&value, sizeof(value));
}

DbStatus IODataSegment::appendInt(uint32_t value) {

    if(m_segMode == DataSegment::READ_ONLY) 
        return DbStatus("IODataSegment::appendInt", DbStatus::DB_READONLY);
    
	return m_segFile->append((char*)&value, sizeof(value));
}

DbStatus IODataSegment::appendLong(uint64_t value) {

    if(m_segMode == DataSegment::READ_ONLY) 
        return DbStatus("IODataSegment::appendLong", DbStatus::DB_READONLY);

	return m_segFile->append((char*)&value, sizeof(value));
}

DbStatus IODataSegment::append(const std::string& data, uint32_t& offset) {

    if(m_segMode == DataSegment::READ_ONLY) 
        return DbStatus("IODataSegment::append", DbStatus::DB_READONLY);

	offset = m_segFile->getNextOffset();
	if(offset + data.size() > m_capacity) return DbStatus("DataSegment::append", DbStatus::DB_SEGMENTOVERFLOW);
	
    //put the length of the payload first
    appendInt(data.size());
	return m_segFile->append(data);
}

DbStatus IODataSegment::append(Dbt& dbt, uint32_t& offset) {

    if(m_segMode == DataSegment::READ_ONLY)
        return DbStatus("IODataSegment::append", DbStatus::DB_READONLY);

    char* buf = dbt.data();
    int n = dbt.getDatalen();

    offset = m_segFile->getNextOffset();
    if(offset + n > m_capacity) return DbStatus("DataSegment::append", DbStatus::DB_SEGMENTOVERFLOW);

    //put the length of the payload first
    appendInt(n);
    return m_segFile->append(buf, n);
}

DbStatus IODataSegment::readShort(uint32_t offset, uint16_t& value) {

	DbStatus ret;
    uint32_t dlen;

    ret = m_segFile->read(offset, sizeof(value), (char*) &value,
            sizeof(value), dlen);

	return ret;
}

DbStatus IODataSegment::readInt(uint32_t offset, uint32_t& value) {

    DbStatus ret;
    uint32_t dlen;

    ret = m_segFile->read(offset, sizeof(value), (char*) &value,
            sizeof(value), dlen);

	return ret;
}

DbStatus IODataSegment::readLong(uint32_t offset, uint64_t& value) {

    DbStatus ret;
    uint32_t dlen;

    ret = m_segFile->read(offset, sizeof(value), (char*) &value,
            sizeof(value), dlen);

	return ret;
}

DbStatus IODataSegment::read(uint32_t offset, uint32_t& size, std::string& result) {

    DbStatus ret;

	uint32_t dataSize = size;

	if(dataSize == 0) {
	
		ret = readInt(offset, dataSize);  if(!ret.isok()) return ret;
        size = dataSize;
	}

	return m_segFile->read(offset+sizeof(uint32_t), dataSize, result);
}


DbStatus IODataSegment::read(uint32_t offset, uint32_t& size, Dbt& dbt) {

    DbStatus ret;

    uint32_t dataSize = size;

    if(dataSize == 0) {
    
        ret = readInt(offset, dataSize);  if(!ret.isok()) return ret;
        size = dataSize;
    }

    //if no buffer is specified, allocate a new buffer,
    if(dbt.empty()) {
        char * buf = new char[dataSize];
        dbt.setBuffer(buf, dataSize, 0, DBT_MALLOC);
    }
    else if(dbt.getSize() < dataSize) return DbStatus("read",
            DbStatus::DB_BUFFERWILLOVERFLOW); 


    uint32_t dlen;
    ret = m_segFile->read(offset+sizeof(uint32_t), dataSize, dbt.data(),
            dbt.getSize(), dlen); 

    dbt.setDatalen(dlen);

    return ret;
}


DbStatus IODataSegment::flush() {

    MetaFile::getInstance()->setDataSize(m_ds->getDsName(),
                                         m_partition->getPartitionName(),
                                         m_segid,
                                         m_dataSize);

    MetaFile::getInstance()->setAlive(m_ds->getDsName(),
                                      m_partition->getPartitionName(),
                                      m_segid,
                                      m_alive);

    return m_segFile->flush();
}


DbStatus IODataSegment::write(uint32_t offset, const std::string& data) {

	uint32_t size = data.size();

	m_segFile->write(offset, (char*)&size, sizeof(uint32_t));
	return m_segFile->write(offset + sizeof(uint32_t), data);
}

uint32_t IODataSegment::getActualDataSize() {

    return m_dataSize;
    /*MetaFile::getInstance()->getDataSize(m_ds->getDsName(),
                                         m_partition->getPartitionName(),
                                         m_segid);*/
}

uint32_t IODataSegment::changeDataSize(int32_t size) {

    if(size < 0 && m_dataSize < (0 - size) ) {
        Logger::log("negative size: %d %d", m_dataSize, size);
    }

    //m_dataSize += size;
    __sync_fetch_and_add(&m_dataSize, size);

    //std::cout<<"change "<<m_segid<<" "<<m_dataSize<<" "<<size<<"\n";

    return m_dataSize;
        
    /*MetaFile::getInstance()->changeDataSize(m_ds->getDsName(),
                                                   m_partition->getPartitionName(),
                                                   m_segid,
                                                   size);*/
}

void IODataSegment::setDataSize(int32_t size) {

    m_dataSize = size;
    m_segFile->setChanged();

    /*
    return MetaFile::getInstance()->setDataSize(m_ds->getDsName(),
                                                m_partition->getPartitionName(),
                                                m_segid,
                                                size);*/ 
}


bool IODataSegment::isAlive() {

    return (m_alive == 1)?true:false;
    /*
    return MetaFile::getInstance()->isAlive(m_ds->getDsName(),
                                            m_partition->getPartitionName(),
                                            m_segid);*/
}

void IODataSegment::setAlive(bool live) {

    m_alive = live?1:0;
    /*
    return MetaFile::getInstance()->setAlive(m_ds->getDsName(),
                                             m_partition->getPartitionName(),
                                             m_segid,
                                             live);*/
}


void IODataSegment::closeWriter() {

    m_segFile->closeWriter();
}

DbStatus IODataSegment::dropCache() {

    return m_segFile->dropCache();
}


void  IODataSegment::notifyCompact() {

    if(!m_notifyCompact && getActualDataSize() <
            getCapacity()*CompactImpl::getInstance()->getValidDataRatio() ) {

        CompactImpl::getInstance()->notify();
        m_notifyCompact = true;
    }
}

